package com.pengheng;

import com.pengheng.bean.WaterSensor;
import com.pengheng.service.ITrafficIndicatorAlgorithm;
import com.pengheng.util.PackageScanner;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Set;

public class Main {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<WaterSensor> streamSource = env.fromElements(
                new WaterSensor("sensor1", 1L, 1),
                new WaterSensor("sensor1", 2L, 2),
                new WaterSensor("sensor2", 2L, 2),
                new WaterSensor("sensor3", 3L, 3)
        );

        OutputTag<WaterSensor> s1 = new OutputTag<>("s1", Types.POJO(WaterSensor.class));
        OutputTag<WaterSensor> s2 = new OutputTag<>("s2", Types.POJO(WaterSensor.class));

        SingleOutputStreamOperator<WaterSensor> ds1 = streamSource.process(new ProcessFunction<WaterSensor, WaterSensor>() {

            @Override
            public void processElement(WaterSensor value, ProcessFunction<WaterSensor, WaterSensor>.Context ctx, Collector<WaterSensor> out) throws Exception {
                if ("s1".equals(value.getId())) {
                    ctx.output(s1, value);
                } else if ("s2".equals(value.getId())) {
                    ctx.output(s2, value);
                } else {
                    out.collect(value);
                }
            }
        });


        Set<Class<?>> classes = PackageScanner.scanPackage("com.pengheng.service.impl");
        for (Class<?> clazz : classes) {
            //如果接口不是弃用 判断是否实现了ITrafficIndicatorAlgorithm接口
            if (!clazz.isAnnotationPresent(Deprecated.class) && ITrafficIndicatorAlgorithm.class.isAssignableFrom(clazz)) {
                // 创建实例并调用方法
                ITrafficIndicatorAlgorithm trafficIndicatorAlgorithm = (ITrafficIndicatorAlgorithm) clazz.newInstance();
                trafficIndicatorAlgorithm.algorithm(ds1);
            }
        }
        // 提交作业执行
        env.execute();
    }
}
